package com.salt.modules.log.listener;

import com.alibaba.fastjson.JSONObject;
import com.salt.modules.log.domain.LoginLog;
import com.salt.modules.log.domain.OperationLog;
import com.salt.modules.log.service.LogService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Objects;

/**
 * @Description
 * @Author salt
 * @Date 2023/4/13
 * @Version 1.0.1
 */

@Component
public class KafkaMessageListener {

    @Autowired
    private LogService logService;

    private static final String OPERATION_LOG_TOPIC = "operation-log";
    private static final String LOGIN_LOG_TOPIC = "login-log";

    //@KafkaListener(topicPartitions = @TopicPartition(topic = "operation-log", partitions = { "0", "1" }), groupId = "salt-log")
    /*@KafkaListener(topics = "operation-log", groupId = "salt-log")
    public void operationLogListen(OperationLog operationLog) {
        logService.save(operationLog);
    }*/

    @Async("threadPoolExecutor")
    @KafkaListener(topics = {"#{'${salt.log.consumer.kafka.topics}'.split(',')}"},
            groupId = "#{'${salt.log.consumer.kafka.group-id}'}")
    public void operationLogListen(ConsumerRecord<String, String> record) {
        if (Objects.equals(record.topic(), OPERATION_LOG_TOPIC)) {
            logService.save(JSONObject.parseObject(record.value(), OperationLog.class));
        } else if (Objects.equals(record.topic(), LOGIN_LOG_TOPIC)) {
            logService.save(JSONObject.parseObject(record.value(), LoginLog.class));
        }
    }


}
